package com.shanshan.wheel.rpc.common.remoting.transport.codec;

import com.shanshan.wheel.rpc.common.compress.GzipCompress;
import com.shanshan.wheel.rpc.common.remoting.constant.RpcConstants;
import com.shanshan.wheel.rpc.common.remoting.constant.RpcMessageType;
import com.shanshan.wheel.rpc.common.remoting.dto.RpcMessage;
import com.shanshan.wheel.rpc.common.remoting.dto.RpcRequest;
import com.shanshan.wheel.rpc.common.remoting.dto.RpcResponse;
import com.shanshan.wheel.rpc.common.serialize.KryoSerializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

import java.util.Arrays;

/**
 * 自定义协议
 * 消息头 16B
 * 4B  magic code（魔法数） 1B version（版本）   4B full length（消息长度[头长 + 消息体长]）
 * 1B messageType（消息类型） 1B codec（序列化类型） 1B compress（压缩类型）  4B  requestId（请求的Id）
 * body（object类型数据）
 */


/**
 * @Description 解码器
 * @ClassName RpcMessageDecoder
 * @Author 山山大魔王
 * @Date 2022/9/4 17:52
 * @Version 1.0
 */
@Slf4j
public class RpcMessageDecoder extends LengthFieldBasedFrameDecoder {

    public RpcMessageDecoder(){
        /**
         * 对于当前协议
         * lengthFieldOffset = magic code + version = 5
         * lengthFieldLength = full length = 4
         * lengthAdjustment = -(magic code + version + full length) = -9
         * 后面需要校验收到的消息是否符合协议因此 initialBytesToStrip = 0
          */
        this(RpcConstants.MAX_FRAME_LENGTH,5,4,-9,0);
    }
    /**
     * @param maxFrameLength 帧的最大长度。如果帧的长度大于这个值，就会抛出TooLongFrameException 。
     * @param lengthFieldOffset 长度字段的偏移量
     * @param lengthFieldLength 长度字段的长度
     * @param lengthAdjustment 添加到长度字段值的补偿值 (因为协议中规定的长度域为整个包的总长，因此需要对前面9个字节进行补偿)
     * @param initialBytesToStrip 从解码帧中剥离的第一个字节数
     */
    public RpcMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        int i1 = in.readableBytes();

        Object decode = super.decode(ctx, in);
        int i2 = in.readableBytes();
        if(decode instanceof ByteBuf){
            ByteBuf frame = (ByteBuf) decode;

            int i = frame.readableBytes();
            if(frame.readableBytes() >= RpcConstants.HEAD_LENGTH){
                try {
                    return decodeFrame(frame);
                } catch (Exception exception) {
                    log.error("解码帧错误！", exception);
                    throw exception;
                } finally {
                    // 释放缓冲区
                    frame.release();
                }
            }
        }
        return decode;
    }

    private Object decodeFrame(ByteBuf in){
        // 校验协议魔数
        checkMagicNumber(in);
        // 校验协议版本
        checkVersion(in);
        // 获取帧的长度
        int fullLength = in.readInt();
        // 获取消息类型
        byte messageType = in.readByte();
        // 获取序列化类型
        byte codecType = in.readByte();
        // 获取压缩类型
        byte compressType = in.readByte();
        // 获取请求id
        int requestId = in.readInt();
        // 更具信息创建消息
        RpcMessage rpcMessage = RpcMessage.builder()
                .messageType(messageType)
                .codec(codecType)
                .compress(compressType)
                .requestId(requestId)
                .requestId(requestId).build();

        // 处理心跳消息
        if(messageType == RpcMessageType.HEARTBEAT_PING_TYPE.getCode()){
            // 设置心跳消息ping
            rpcMessage.setData("PING");
        }

        if(messageType == RpcMessageType.HEARTBEAT_PONG_TYPE.getCode()){
            // 设置心跳消息pong
            rpcMessage.setData("PONG");
        }
        // 求出消息内容长度
        int bodyLength = fullLength - RpcConstants.HEAD_LENGTH;
        if(bodyLength > 0){
            // 读取消息内容
            byte[] bs = new byte[bodyLength];
            in.readBytes(bs);
            // 数据解压缩
            GzipCompress gzipCompress = new GzipCompress();
            byte[] decompress = gzipCompress.decompress(bs);
            // 定义序列化器
            KryoSerializer kryoSerializer = new KryoSerializer();

            if(messageType == RpcMessageType.REQUEST_TYPE.getCode()){
                // 反序列化
                RpcRequest rpcRequest = kryoSerializer.deserialize(decompress, RpcRequest.class);
                // 设置请求数据
                rpcMessage.setData(rpcRequest);
            }else {
                // 反序列化
                RpcResponse rpcResponse = kryoSerializer.deserialize(decompress, RpcResponse.class);
                // 设置响应数据
                rpcMessage.setData(rpcResponse);
            }
        }

        return rpcMessage;
    }

    private void checkVersion(ByteBuf in){
        // 读取一个字节
        byte version = in.readByte();
        if(version != RpcConstants.VERSION){
            throw new RuntimeException("版本不兼容" + version);
        }
    }

    private void checkMagicNumber(ByteBuf in){

        int len = RpcConstants.MAGIC_NUMBER.length;
        byte[] temp = new byte[len];
        in.readBytes(temp);

        for(int i = 0; i < len; ++i){
            if(temp[i] != RpcConstants.MAGIC_NUMBER[i]){
                throw new IllegalArgumentException("未知的魔法密码："+ Arrays.toString(temp));
            }
        }
    }
}
